package com.zhengb.flink.study.datastream;

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.HashMap;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 实时订单统计demo
 *
 * @author zhengbo
 * @date 2019/11/26
 */
@Slf4j
public class GroupedProcessingTimeWindowExample {

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class SellEntity {

        private String type;

        private Integer sellNum;
    }

    /**
     * 模拟订单生成数据源
     */
    private static class DataSource extends RichParallelSourceFunction<SellEntity> {

        private volatile static boolean isRunning = true;

        @Override
        public void run(SourceContext<SellEntity> context) throws Exception {

            ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
            while (isRunning) {
                Thread.sleep((getRuntimeContext().getIndexOfThisSubtask() + 1) * 1000 * 5);
                //商品类别
                String type = "类别" + (char) ('A' + threadLocalRandom.nextInt(3));
                //商品数量
                int sellNum = threadLocalRandom.nextInt(10) + 1;

                log.info("Emits:{},{}\t", type, sellNum);

                context.collect(new SellEntity(type, sellNum));
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataStreamSource<SellEntity> streamSource = env.addSource(new DataSource());

        KeyedStream<SellEntity, Tuple> keyedStream = streamSource.keyBy("type");

        keyedStream.sum("sellNum").keyBy(new KeySelector<SellEntity, Object>() {
            @Override
            public Object getKey(SellEntity value) throws Exception {
                //返回商品value.fo种类  则按商品分类统计
                //log.info("keyBY:{}", value.f0);
                //return value.f0;

                //返回空字符串  对商品不分组 所有商品聚合在一起统计
                return "";
            }
        }).fold(new HashMap<>(), new FoldFunction<SellEntity, HashMap<String, Integer>>() {
            @Override
            public HashMap<String, Integer> fold(HashMap<String, Integer> accumulator, SellEntity value) throws Exception {

                accumulator.put(value.getType(), value.getSellNum());
                return accumulator;
            }
        }).addSink(new SinkFunction<HashMap<String, Integer>>() {

            @Override
            public void invoke(HashMap<String, Integer> value, Context context) throws Exception {
                //每个类型商品成交量
                log.info("每个类型商品成交量:{}", JSONObject.toJSONString(value));
                //商品总成交量
                log.info("商品总成交量:{}", value.values().stream().mapToInt(x -> x).sum());
            }
        });

        env.execute("processingShop");
    }

}
